#ifdef __cplusplus
extern "C"{
#endif

#include "xs_def.h"
#include "xs_malloc.h"
#include "xs_log.h"
#include "xs_io.h"
#include "xs_linux.h"
#include "xs_vector.h"

#ifdef WIN32
int xs_socketpair(int af, int type, int protocol, XSOCKET socks[])
{
    protocol = protocol;
    XSOCKET listener, connector, acceptor;
    struct sockaddr_in listen_addr, connect_addr;
    int size = sizeof(connect_addr);

    listener = socket(af, type, 0);
    if(listener == (XSOCKET)SOCKET_ERROR)
        return -1;

    memset(&listen_addr, 0, sizeof(listen_addr));
    listen_addr.sin_family = AF_INET;
    listen_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    listen_addr.sin_port = 0;

    if(bind(listener, (struct sockaddr*)&listen_addr, sizeof(listen_addr)) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }

    if(listen(listener, 1) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }


    if(getsockname(listener, (struct sockaddr*)&connect_addr, &size) == -1)
    {
        xs_close_socket(listener);
        return -1;
    }
    connector = socket(af, type, 0);
    if(connector == (XSOCKET)SOCKET_ERROR)
    {
        xs_close_socket(listener);
        return -1;
    }

    if(connect(connector, (struct sockaddr*)&connect_addr, size) == -1)
    {
        xs_close_socket(connector);
        xs_close_socket(listener);
        return -1;
    }

    size = sizeof(listen_addr);
    acceptor = accept(listener, (struct sockaddr*)&listen_addr, &size);
    if(acceptor == (XSOCKET)SOCKET_ERROR)
    {
        xs_close_socket(connector);
        xs_close_socket(listener);
        return -1;
    }

    socks[0] = connector;
    socks[1] = acceptor;

    xs_logd("socketpair=%d, %d", socks[0], socks[1]);

    return 0;
}

/* use select do epoll */
typedef struct xs_epoll_event_t
{
    XSOCKET fd;
    struct epoll_event ev;
    xs_list_t lnode;
    struct rb_node rb;
} xs_epoll_event_t;

typedef struct xs_epoll_evs_t
{
    xs_list_t ready;
    xs_list_t go;
    struct rb_root tree;
} xs_epoll_evs_t;

pthread_mutex_t g_epoll_mutex = PTHREAD_MUTEX_INITIALIZER;
xs_epoll_evs_t* g_epoll_evs = NULL;
#define xs_epoll_lock() pthread_mutex_lock(&g_epoll_mutex)
#define xs_epoll_unlock() pthread_mutex_unlock(&g_epoll_mutex)
#define xs_epoll_find(_tree, _key) rb_find(_tree, (const void*)(intptr_t)_key)

int xs_epoll_evs_cmp(const struct rb_node* node, const void* ptr)
{
    xs_epoll_event_t* ev = rb_entry(node, xs_epoll_event_t, rb);
 //   xs_logd("%d, key=%d", ev->fd, (XSOCKET)(intptr_t)ptr);
    return ev->fd - (XSOCKET)(intptr_t)ptr;
}

xs_epoll_event_t* xs_epoll_event_create(XSOCKET fd, struct epoll_event* ev)
{
    xs_epoll_event_t* xsev = xs_malloc(sizeof(xs_epoll_event_t));
    memcpy(&xsev->ev, ev, sizeof(*ev));
    xsev->fd = fd;
    rb_init_node(&xsev->rb);
    xs_list_init(&xsev->lnode);
    return xsev;
}

int xs_epoll_ctl(int epollfd, int method, XSOCKET fd, struct epoll_event* ptr)
{
    int ret = 0;
    struct rb_node* node;
    xs_epoll_event_t* xsev;

    epollfd = epollfd;

    xs_epoll_lock();
    if(g_epoll_evs == NULL)
    {
        g_epoll_evs = xs_malloc(sizeof(xs_epoll_evs_t));
        xs_list_init(&g_epoll_evs->go);
        xs_list_init(&g_epoll_evs->ready);
        rb_init(&g_epoll_evs->tree, xs_epoll_evs_cmp);
    }
    node = xs_epoll_find(&g_epoll_evs->tree, fd);

    if(node)
    {
        xsev = xs_entry(node, xs_epoll_event_t, rb);
        switch (method)
        {
            case EPOLL_CTL_DEL:
                rb_erase(&xsev->rb, &g_epoll_evs->tree);
                xs_list_del(&xsev->lnode);
                goto RET;

            case EPOLL_CTL_MOD:
                memcpy(&xsev->ev, ptr, sizeof(*ptr));
                xs_list_move_tail(&g_epoll_evs->ready, &xsev->lnode);
                goto RET;

            case EPOLL_CTL_ADD:
                /* already exist, can not add again */
                xs_logd("already exist fd=%d", xsev->fd);
                ret = -1;
                goto RET;
        }
    }
    else
    {
        switch (method)
        {
            /* not exist, can not del and mod */
            case EPOLL_CTL_DEL:
            case EPOLL_CTL_MOD:
                ret = -1;
                goto RET;
            case EPOLL_CTL_ADD:
                /* add the event in ready list */
                xsev = xs_epoll_event_create(fd, ptr);
                rb_insert(&xsev->rb, &g_epoll_evs->tree, (void*)(intptr_t)fd);
                xs_list_add_tail(&g_epoll_evs->ready, &xsev->lnode);
                goto RET;
        }
    }
RET:
    xs_epoll_unlock();
    return ret;
}

int xs_epoll_wait(int epollfd, struct epoll_event* ev_ret, int count, int t)
{
    int i;
    xs_list_t* node;
    xs_epoll_event_t* ev;
    struct timeval tv;
    fd_set r;
    fd_set w;
    FD_ZERO(&r);
    FD_ZERO(&w);

    xs_epoll_lock();
    if(!xs_list_empty(&g_epoll_evs->go))
        goto RET;

    epollfd = epollfd;
    node = g_epoll_evs->ready.next;
    while(node != &g_epoll_evs->ready)
    {
        ev = xs_entry(node, xs_epoll_event_t, lnode);
        node = node->next;
        if(ev->ev.events & EPOLLIN)
        {
            FD_SET(ev->fd, &r);
        }
        if(ev->ev.events & EPOLLOUT)
        {
            FD_SET(ev->fd, &w);
        }
    } 
    xs_epoll_unlock();


    tv.tv_sec = t/1000;
    tv.tv_usec = t%1000*1000;

    int ret = select(1024, &r, &w, NULL, &tv);
    if(ret <= 0)
        return ret;

    xs_epoll_lock();

    if(!xs_list_empty(&g_epoll_evs->ready))
    {
        node = g_epoll_evs->ready.next;
        while(node != &g_epoll_evs->ready)
        {
            ev = xs_list_entry(node, xs_epoll_event_t, lnode);
            node = node->next;
            if(FD_ISSET(ev->fd, &r) || FD_ISSET(ev->fd, &w))
            {
                xs_list_move_tail(&g_epoll_evs->go, &ev->lnode);
            }
        }
    }

RET:
    for(i=0; i<count; ++i)
    {
        if(xs_list_empty(&g_epoll_evs->go))
            break;
        
        ev = xs_list_first_entry(&g_epoll_evs->go, xs_epoll_event_t, lnode);
        xs_list_del(&ev->lnode);
        memcpy(&ev_ret[i], &ev->ev, sizeof(struct epoll_event));
    }
    xs_epoll_unlock();
    return i;
}

int xs_epoll_create(int __c)
{
    g_epoll_evs = xs_malloc(sizeof(xs_epoll_evs_t));
    xs_list_init(&g_epoll_evs->go);
    xs_list_init(&g_epoll_evs->ready);
    rb_init(&g_epoll_evs->tree, xs_epoll_evs_cmp);
    __c = __c
    return 0;
}

int xs_epoll_close(int fd)
{
    fd = fd;
    return 0;
}

#else		/* End ifdef __WIN32 and Start of Linux */

#include "xs_epoll.h"

int xs_socketpair(int __a, int __type, int __pro, XSOCKET __socks [ 2 ])
{
	int __result__ = -1;

	__result__ = (socketpair(__a, __type, __pro, __socks));
	if(__result__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
	
	return EXIT_OK;
}
int xs_epoll_create(int __c)
{
	XSOCKET __efd__ = -1;

	__efd__ = epoll_create(__c);
	if(__efd__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
		
	return (__efd__);
}
int xs_epoll_ctl(XSOCKET __efd, int __method, XSOCKET __fd, struct epoll_event* __p)
{
	int __result__ = -1;

	__result__ = (epoll_ctl(__efd, __method, __fd, __p) < 0);
	if(__result__ < 0){
		xs_dump_error_str(errno);
		return EXIT_ERROR;
	}
	
	return EXIT_OK;
}
int xs_epoll_wait(XSOCKET __efd, struct epoll_event* __ev, int __c, int __tw)
{
	int __num__ = -1;

	__num__ = epoll_wait(__efd, __ev, __c, __tw);
	if(__num__ < 0)
		__num__ = EXIT_ERROR;

	return (__num__);
}
int xs_epoll_close(XSOCKET __f)
{
	return xs_release_socket(__f);
}

#endif

/*
	Add fd to epollfd
	Register the target file descriptor fd on the epoll instance referred to by the file descriptor epfd 
	and associate the event with the internal file linked to fd.
*/
int xs_epoll_add(int __epollfd, int __fd, int __mod)
{
	xs_epoll_event __xs_ev__;
	
	__xs_ev__.events = __mod;
	__xs_ev__.data.fd = __fd;

	if(xs_epoll_ctl(__epollfd, EPOLL_CTL_ADD, __fd, &__xs_ev__) < 0)
	        return EXIT_ERROR;  

	return EXIT_OK;
}

/* 	
	Remove  (deregister) the target file descriptor fd from the epoll instance referred to by epollfd.  
	The event is ignored and can be NULL. 
	In kernel versions before 2.6.9, the EPOLL_CTL_DEL operation required a non-NULL pointer in event
	Since Linux 2.6.9, event can be specified as NULL when using EPOLL_CTL_DEL. 
*/
int xs_epoll_del(int __epollfd, int __fd)
{
	xs_epoll_event __xs_ev__;

	/* We should better set event not NULL */
	__xs_ev__.events = __xs_ev__.events;
	__xs_ev__.data.fd = __fd;

	if(xs_epoll_ctl(__epollfd, EPOLL_CTL_DEL, __fd, &__xs_ev__) < 0)
	        return EXIT_ERROR;  

	return EXIT_OK;
}

/* Change the event associated with the target file descriptor __fd. */
int xs_epoll_mode(int __epollfd, int __fd, int __newmod)
{
	xs_epoll_event __xs_ev__;

	__xs_ev__.events = __newmod;
	__xs_ev__.data.fd = __fd;

	if(xs_epoll_ctl(__epollfd, EPOLL_CTL_MOD, __fd, &__xs_ev__) < 0)
	        return EXIT_ERROR;  

	return EXIT_OK;
}

/* 
	xs_epoll_init: creates  an epoll "instance", requesting the kernel to allocate an 
	event backing store dimensioned for size descriptors.
	Since Linux 2.6.8, the size argument is unused, but must be greater than zero.
	(The kernel  dynamically  sizes  the  required  data
       structures without needing this initial hint.)
*/
int xs_epoll_init(int size)
{
	int epollfd = -1;
	
	if(size == 0)
		epollfd = xs_epoll_create(EPOLL_DEFAULT_SIZE);
	else epollfd = xs_epoll_create(size);
	

	if(epollfd < 0) return EXIT_ERROR;

	xs_logd("epoll create success -> fd : %d", epollfd);

	return epollfd;
}



#include "xs_network.h"
#define __VECTOR__
typedef struct block_queue  
{  
#ifdef __VECTOR__ 
	vector queue;
#else
	int queue[xs_defult_thread_size];
	long size;    
#endif
	xs_pthread_cond_t cond;  
	xs_pthread_mutex_t mutex;  
}block_queue_t;  
block_queue_t *bq;  

typedef struct block_queue_param  
{  
	void* func;  
	void* queue;  	/* Point to block queue structure */
}block_queue_param_t;  
block_queue_param_t bqp; 

int g_xs_thread_count = xs_defult_thread_size;
#define BUFFER_SIZE 		1024
#include <sys/resource.h>
int g_epoll_fd = -1;
xs_epoll_event xs_ev, xs_events[EPOLL_DEFAULT_SIZE];
int g_epoll_size = EPOLL_DEFAULT_SIZE;
int g_serv_fd = -1;

static int xs_queue_init( block_queue_t *__q)
{
	if(__q == NULL) return (-1);
	
#ifdef __VECTOR__
	__q->queue = vector_init(xs_defult_thread_size);
#else
	__q->size = 0;
#endif

	xs_pthread_cond_init(&(__q->cond), NULL);  
	xs_pthread_mutex_init(&(__q->mutex), NULL);  

	return 0;
}

block_queue_t *xs_epoll_queue_create( void )
{
	block_queue_t *__q;

	__q = xs_malloc(sizeof(block_queue_t));

	assert(__q);

	return ((xs_queue_init(__q) == 0) ? __q : NULL);
}
static inline void xs_network_epoll_loop(void* data)
{
	 int socket;

      socket = *(int *)data;

	xs_logd("%d !\n", socket);
	char buffer[BUFFER_SIZE];
	xs_pthread_t id = pthread_self();
	xs_logd("thread id is: %ld", id);

	/* We only send what recevied just now */
	int   length = xs_net_recv(socket, buffer, BUFFER_SIZE);
	if(length){
		xs_net_send(socket, buffer, strlen(buffer));
		memset(buffer, 0, BUFFER_SIZE);
	}
}

void *xs_handle_queue(void *param)  
{  
	void(* func)(void* );  
	int fd;

	block_queue_t* bque = ((block_queue_param_t*)param)->queue;  
	func = ((block_queue_param_t*)param)->func; 

	xs_pthread_cond_init(&bque->cond,  NULL); 
	xs_pthread_mutex_init(&bque->mutex,  NULL);
	
	for(;;)  
	{  
		if(xs_pthread_mutex_lock(&bque->mutex) == EXIT_OK) {
			
			xs_pthread_cond_wait(&bque->cond, &bque->mutex);  
#ifdef __VECTOR__
			if(bque->queue->active == 0){
				xs_pthread_mutex_unlock(&bque->mutex);
				continue;
			}else {
				fd = *(int *)(vector_lookup(bque->queue, 0));
			}
#else
			if(bque->size==0)  {  
				xs_pthread_mutex_unlock(&bque->mutex);
				continue;
			}else {  	
				
				int i;
				fd = bque->queue[0];  

				for(i = 0; i < bque->size - 1; ++i)  
					bque->queue[i] = bque->queue[i + 1];  
				
				bque->queue[bque->size-1] = 0;  
				bque->size--;  
			}  
#endif
		    	xs_pthread_mutex_unlock(&bque->mutex);  
		}
		
		func((void *)&fd);  
	}  
}  

int xs_init_threads( void )  
{  
	int i = 0, ret;  
	xs_pthread_t child_thread[g_xs_thread_count];  
	xs_pthread_attr_t child_thread_attr[g_xs_thread_count];  

	bqp.func = (void*)xs_network_epoll_loop; 
	bqp.queue = (void *)bq;

	for( i = 0; i < g_xs_thread_count; ++i)  {  
		ret = xs_pthread_attr_init(&child_thread_attr[i]);  
		if(ret != 0) xs_logd("error to init attr !\n");
		pthread_attr_setdetachstate(&child_thread_attr[i], PTHREAD_CREATE_DETACHED);  
		if( pthread_create(&child_thread[i],
			&child_thread_attr[i], xs_handle_queue, (void *)&bqp) < 0 ) {  
				xs_logd("pthread_create Failed : %s - %m\n",strerror(errno));  
				return 1;  
		}  
    	}  
  	return 0;
}  
int xs_init_server(const char *name, short int port)
{
	struct rlimit rt;  
	int server_socket = -1;
	
	server_socket = xs_create_server(name, port);

	rt.rlim_max = rt.rlim_cur = g_epoll_size;  
	if (setrlimit(RLIMIT_NOFILE, &rt) == -1) {  
		xs_logd("setrlimit - %m");  
		exit(1);  
	}  

	return server_socket;  
}

static void xs_insert_queue(block_queue_t *bque, int *fd)  
{  
	xs_pthread_mutex_lock(&bque->mutex);  
	
#ifdef __VECTOR__
	vector_set(bque->queue, fd);
#else
	if(bque->size == g_xs_thread_count)  
		return;  

	bque->queue[bque->size] = *fd;  
	bque->size++;
	if(bque->size > g_xs_thread_count)  {  
		fprintf(stderr,"Queue size over folow.%ld", bque->size);  
		exit (1);  
	}  
#endif

	xs_pthread_cond_signal(&bque->cond);  
	xs_pthread_mutex_unlock(&bque->mutex);  

} 

static inline void xs_handler(void* fd)  
{  
	printf("handler:fd => %d\n", *(int *)(fd));  
	xs_insert_queue(bq, fd);

}  
int xs_epoll_entry()
{
	int nfds, n;
	
	g_serv_fd = xs_init_server(NULL, 11111);
	xs_logd("server thread [FD:%d] is ready for ...", g_serv_fd);

	
   	bq = xs_epoll_queue_create();
	assert(bq);
	 if(xs_init_threads() == 0)  
        	xs_logd("Threads ready to use !");  

	g_epoll_fd = xs_epoll_init(g_epoll_size);
	xs_epoll_add(g_epoll_fd, g_serv_fd, EPOLLIN | EPOLLET);

	 for(;;) {  
	        struct sockaddr_in local;  
	        socklen_t length = sizeof(local);  
	        int client = -1;  
	          
	        nfds = xs_epoll_wait(g_epoll_fd, xs_events, EPOLL_DEFAULT_SIZE, epoll_wait_indefinite);  

	        for(n = 0; n < nfds; ++n) {  

			if(xs_events[n].data.fd == g_serv_fd) {  
				client = xs_net_accept(g_serv_fd, (struct sockaddr *)&local, &length);  
				if(client < 0) {  
					xs_logd("%s", strerror(errno));  
					continue;  
				}  else {
					xs_logd("add socket pool : %d", client);
					set_nonblocking(client);  
					xs_epoll_add(g_epoll_fd, client, EPOLLIN | EPOLLOUT | EPOLLET);
					client = -1;
				}
			}  
			else  /* It's a client fd that needed to process */
				xs_handler((void *)&xs_events[n].data.fd);
	        }  
    	}  

	xs_pthread_mutex_destroy(&bq->mutex);
	xs_pthread_cond_destroy(&bq->cond);
	xs_close_socket(g_serv_fd); 
	xs_epoll_close(g_epoll_fd);
#ifdef __VECTOR__
	vector_free(bq->queue);
#endif

	xs_free(bq);

}

#ifdef __cplusplus
}
#endif
